package com.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.common.GlobalConfig;
import com.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2022/1/2 22:36
 */

//TODO: 数据流 web/app -> Nginx -> Springboot  -> Kafka(ods)  ->  FlinkApp  -> Kafka(dwd)
//TODO: 程序 mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO: 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //TODO:Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
            /*//开启CK并指定状态后端为FS menory fs rocksdb
            env.setStateBackend(new FsStateBackend("hdfs://192.168.88.109:9820/gmall-flink/ck"));
            //开启 Checkpoint,每隔 5 秒钟做一次 CK
            env.enableCheckpointing(5000L);
            //指定 CK 的一致性语义
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(10000L);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
            //设置任务关闭的时候保留最后一次 CK 数据
            env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            //指定从 CK 自动重启策略
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000L));
            //设置访问 HDFS 的用户名
            System.setProperty("HADOOP_USER_NAME", "root");*/

        //TODO: 2.消费 ods_base_log 主题数据创建流
        String sourceTopic = "ods_base_log";
        String groupId="base_log_app"+ GlobalConfig.NUMBER;
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO: 3.将每行数据转换为JSON对象
        /*kafkaDS.map( line ->{
            return JSON.parseObject(line);
        })*/
        /*kafkaDS.map(JSON::parseObject)*/
        OutputTag<String> dirtyTag = new OutputTag<String>("Dirty") {};
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    //发生异常，将数据写入侧输出流
                    context.output(dirtyTag, s);

                }
            }
        });

        //脏数据打印
        jsonObjDS.getSideOutput(dirtyTag).print("dirty>>>>>>>>");

        //TODO: 4.新老用户校验  状态编程
        SingleOutputStreamOperator<JSONObject> jsonObjNewFlag = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObject) throws Exception {
                        //获取数据中的”is_new“标记
                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
                        //判断isNew标记是否为”1“
                        if (isNew.equals("1")) {
                            //获取状态数据
                            String state = valueState.value();

                            if (state != null) {
                                //修改isNew标记
                                jsonObject.getJSONObject("common").put("is_new", "0");
                            } else {
                                valueState.update("1");
                            }
                        }
                        return jsonObject;

                    }
                });

        //TODO: 5.分流  侧输出流  页面：主流  启动：侧输出流  曝光：侧输出流
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        OutputTag<String> displayTag = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> pageDS = jsonObjNewFlag.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {

                //获取启动日志字段
                String start = jsonObject.getString("start");
                if (start != null && start.length() > 0) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    //将数据写入页面日志主流
                    collector.collect(jsonObject.toJSONString());

                    //取出数据中的曝光数据
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        //获取页面id
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");

                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);

                            //添加页面id
                            display.put("page_id", pageId);

                            //将数据写入到曝光侧输出流
                            context.output(displayTag, display.toJSONString());

                        }
                    }

                }
            }
        });


        //TODO: 6.提取侧输出流
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);

        //TODO: 7.将3个流进行打印输出到对应的  kafka  主题中
        System.out.println("任务开始....................");

        startDS.print("start>>>>>>>>");
        displayDS.print("display>>>>>>>>");
        pageDS.print("page>>>>>>>>");

        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));
        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));


        //TODO: 8.启动任务
        env.execute("BaseLogApp");



    }
}
